package com.joker;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

import static com.joker.ZookeeperConstant.ZK_ADDRESS;

public class ZooKeeperWatcherExample {
    private ZooKeeper zk;

    public void watchNode(String path) throws KeeperException, InterruptedException, IOException {
        CountDownLatch connectedSignal = new CountDownLatch(1);

        zk = new ZooKeeper(ZK_ADDRESS, 5000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                if (event.getType() == Event.EventType.NodeDataChanged) {
                    System.out.println("节点数据发生变化: " + event.getPath());
                    try {
                        // 重新注册watcher
                        byte[] data = zk.getData(path, this, null);
                        System.out.println("新数据: " + new String(data));
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                } else if (event.getState() == Event.KeeperState.SyncConnected) {
                    connectedSignal.countDown();
                }
            }
        });

        connectedSignal.await();

        System.out.println("连接成功!");

        Stat stat = zk.exists(path, false);
        if (stat == null) {
            System.out.println("节点不存在: " + path);
            // 创建节点
            zk.create(path, "处事".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            System.out.println("节点已创建: " + path);
        }

        // 初始获取数据并设置watcher
        byte[] data = zk.getData(path, true, null);
        System.out.println("初始数据: " + new String(data));

        System.out.println("开始监听节点: " + path);
        // 开启一个异步线程往里面发送数据
        new Thread(() -> {
            while (true) {
                try {
                    System.out.println("正在更新数据...");
                    zk.setData(path, ("当前时间: " + System.currentTimeMillis()).getBytes(), -1);
                    Thread.sleep(1000);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    public static void main(String[] args) throws Exception {
        ZooKeeperWatcherExample example = new ZooKeeperWatcherExample();
        example.watchNode("/test-watch1");

        // 保持程序运行以持续监听
        Thread.sleep(60000);
    }
}